Skip to content

feat: job maintenance API (remove / drain / clean / obliterate)#146

Merged
jotarios merged 8 commits into
mainfrom
feat/job-maintenance-api
May 23, 2026
Merged

feat: job maintenance API (remove / drain / clean / obliterate)#146
jotarios merged 8 commits into
mainfrom
feat/job-maintenance-api

Conversation

@jotarios

Copy link
Copy Markdown
Owner

Summary

Adds the job maintenance APIremove, drain, clean, obliterate — across the engine and both shims. The Node Queue.remove / drain / clean / obliterate methods previously threw NotSupportedError; the Python Queue lacked them entirely. All four now route through new engine surface.

Streams-based semantics, adapted (not copied) from list-based queues:

  • remove(jobId) — delete one job from every surface it could live on: the delayed ZSET (+ didx / dlid side-indexes), a waiting or active main-stream entry, the DLQ, and the result key. Idempotent — a missing id returns an all-false report, not an error.
  • drain(delayed?) — clear waiting stream entries (not in any consumer-group PEL) plus, by default, the delayed ZSET. In-flight jobs are left running.
  • clean(grace, limit, state) — age- + state-filtered bulk delete for completed / failed / delayed / waiting, capped at limit, returning the removed job ids.
  • obliterate() — tear the whole {chasqui:<queue>} keyspace down via a batched SCAN + UNLINK.

What shipped

Engine (chasquimq/src/producer/maintenance.rs, new) — Producer::remove / drain / clean / obliterate, plus RemovalReport / DrainOptions. Three new atomic Lua scripts (REMOVE_STREAM_ENTRY, DRAIN_STREAM, CLEAN_STREAM) follow the existing EVALSHANOSCRIPTEVAL self-heal pattern; the delayed branch reuses CANCEL_DELAYED_SCRIPT verbatim. All scans bounded so no call blocks Redis. Off the hot path — zero diff to consumer/, ack.rs, promoter.rs, scheduler.rs, and no add* method touched.

Node shim — four new napi producer bindings + un-stubbed Queue.remove / drain / clean / obliterate; Queue.removeReport exposes the per-surface RemovalReport.

Python shim — four PyO3 bindings + Queue.remove / remove_report / drain / clean / obliterate (previously absent); _native.pyi updated.

CLIchasqui clean <queue> --state <s> --grace-ms <ms> --limit <n> and chasqui obliterate <queue>, both gated behind a y/N confirm unless --yes.

Docsdocs/engine.md (new section), docs/history.md (slice entry), both shim READMEs (symmetric sections), root README feature row, reference/{node-api,python-api,rust-api,cli}.md, a new guides/clean-and-obliterate.mdx registered in the sidebar. Stale claims fixed inline: the "drain/clean/obliterate are stubbed" note, the "engine doesn't implement most job states" note, and the DLQ guide's "no drop command" line.

Notable design call

clean(delayed) ages by the job's created_at_ms (when scheduled), not the ZSET run-at score — a future-scheduled job is otherwise impossible to clean deterministically. "Drop delayed jobs created more than grace ago" is a stable, testable semantic.

Bugs caught in review

A diff review surfaced two correctness bugs, both fixed with regression tests:

  • drain multi-pass termination — the pass loop stopped on a partial scan page, not on zero. With active entries interleaved near the front of the stream, a pass legitimately deletes fewer than a full page while thousands of waiting jobs remain further back. Now stops only when a pass deletes zero.
  • clean(waiting) PEL skip — the candidate scan walked the main stream, which mixes waiting and active entries, so an old-enough in-flight job could be swept into the delete set. clean(waiting) now subtracts the consumer-group PEL.

Testing

  • Engine: 21 live-Redis integration tests (every state, idempotency, active-survives-drain/clean, obliterate-then-reuse, multi-pass-past-scan-page).
  • Node: 13 vitest cases. Python: 14 pytest cases.
  • Full suites green: 356 Rust (38 suites), 185 Node, 149 Python. cargo fmt / clippy -D warnings / tsc / prettier all clean.

Benchmark

Same-host run (load avg ~3.5, contended): queue-add-bulk 179,572 jobs/s, worker-concurrent 108,701 jobs/s — within the host-load-noise envelope vs the chasquimq-1.0.md baseline. The maintenance API adds zero code to the producer/consumer hot paths (verified: empty diff to consumer/ and add*), so the small deltas are host contention, not a regression — consistent with the host-load gate in benchmarks/README.md.

jotarios added 8 commits May 22, 2026 16:59
…rate

Add four operator/SDK surfaces on Producer for tearing jobs (or a whole
queue) down, adapting BullMQ-shaped semantics to a Redis Streams engine:

- remove(job_id, group): delete one job everywhere it could live — the
  delayed ZSET + didx/dlid side-indexes, a waiting or active main-stream
  entry, the DLQ, and the result key. Idempotent; returns a RemovalReport
  flagging which surfaces actually held it.
- drain(group, opts): clear waiting stream entries (not in any PEL) plus,
  by default, the delayed ZSET. In-flight jobs are left running.
- clean(group, grace_ms, limit, state): age- + state-filtered bulk delete
  for completed/failed/delayed/waiting. Age basis is the stream entry id
  for waiting/failed, created_at_ms for delayed; ignored for completed.
- obliterate(group): nuke the entire {chasqui:<queue>} keyspace via a
  batched SCAN + UNLINK (async reclaim, never blocks Redis).

Three new atomic Lua scripts (REMOVE_STREAM_ENTRY, DRAIN_STREAM,
CLEAN_STREAM) follow the existing EVALSHA -> NOSCRIPT -> EVAL self-heal
pattern; the delayed branch reuses CANCEL_DELAYED_SCRIPT verbatim. All
scans are bounded by MAINTENANCE_SCAN_PAGE so no call can block Redis.

19 integration tests against live Redis cover every state and the
idempotent / active-survives-drain / obliterate-then-reuse edges.
Un-stub Queue.remove / drain / clean / obliterate — previously they threw
NotSupportedError. All four now route through new native producer
bindings to the engine maintenance API:

- remove(jobId): deletes a job from every surface; returns the count of
  surfaces hit. Queue.removeReport gives the per-surface RemovalReport.
- drain(delayed = true): clears waiting + (optionally) delayed jobs;
  returns the count removed.
- clean(grace, limit, type): age- + state-filtered bulk delete; returns
  the removed job ids. type defaults to "completed".
- obliterate(): tears the whole queue keyspace down; returns the key
  count removed.

RemovalReport is re-exported from the package entry point. 13 shim tests
cover every state plus the idempotent / obliterate-then-reuse edges.
Add Queue.remove / drain / clean / obliterate to the Python shim,
mirroring the Node surface. New PyO3 producer bindings route to the
engine maintenance API:

- remove(job_id): deletes a job from every surface; returns the count of
  surfaces hit. Queue.remove_report gives the per-surface dict.
- drain(delayed=True): clears waiting + (optionally) delayed jobs;
  returns the count removed.
- clean(grace_ms, limit, state): age- + state-filtered bulk delete;
  returns the removed job ids. state defaults to "completed".
- obliterate(): tears the whole queue keyspace down; returns the key
  count removed.

_native.pyi updated with the four producer method stubs. 14 shim tests
mirror the Node maintenance suite.
Two operator subcommands over the engine maintenance API:

- `chasqui clean <queue> --state <s> --grace-ms <ms> --limit <n>` —
  age- and state-filtered bulk delete; prints the removed job ids.
- `chasqui obliterate <queue>` — tears the whole queue keyspace down.

Both are destructive and gated behind an interactive y/N confirm unless
`--yes` is passed, matching the existing `chasqui dlq replay` pattern.
Sync every doc surface for remove / drain / clean / obliterate:

- docs/engine.md: new "Job maintenance" section.
- docs/history.md: job-maintenance slice changelog entry.
- README.md: new feature-table row.
- Node + Python READMEs: symmetric "Maintenance" sections.
- reference/{node-api,python-api,rust-api,cli}.md: full method docs;
  drop the stale "drain/clean/obliterate are stubbed" note; correct the
  stale "engine doesn't implement most job states" note.
- new guides/clean-and-obliterate.mdx, registered in the sidebar.
- replay-the-dlq.mdx: "Drop instead" now points at `chasqui clean`
  rather than a raw redis-cli DEL.
Two correctness bugs caught reviewing the maintenance diff:

- drain: the pass loop stopped on a partial scan page (`pass < page`),
  not on zero. When Active (PEL) entries are interleaved near the front
  of the stream, a pass legitimately deletes fewer than a full page
  while thousands of waiting jobs remain further back — the old
  condition left them undrained. Stop only when a pass deletes zero;
  add a hard iteration cap as a belt-and-braces guard.
- clean(Waiting): the candidate scan walked `XRANGE - +` over the main
  stream, which mixes waiting and Active entries, so an old-enough
  in-flight job could be swept into the delete set. clean(Waiting) now
  subtracts the consumer-group PEL (via XPENDING) before deleting; the
  DLQ path is unaffected (a plain stream, no group).

Two new live-Redis regression tests:
`drain_multi_pass_clears_all_waiting_past_scan_page` (2500 waiting jobs
past the 1024 scan-page cap) and `clean_waiting_leaves_active_jobs`.
The job-picked-up / DLQ-populated wait loops used a 10s budget, which
can be exceeded when all workspace test suites run in parallel against
one shared Redis. Widen the five wait_until budgets to 30s and the
blocking-handler sleeps to 120s so they comfortably outlive the wait.
No logic change — robustness only.
@jotarios jotarios merged commit 9ad8be4 into main May 23, 2026
26 checks passed
@jotarios jotarios deleted the feat/job-maintenance-api branch May 23, 2026 03:36
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant